package jk.fox.transport.netty.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.AttributeKey;
import jk.fox.common.RpcDict;
import jk.fox.common.exception.runtime.RpcRequestTimeoutException;
import jk.fox.common.payload.message.MessageDispatcher;
import jk.fox.common.payload.request.Request;
import jk.fox.common.payload.response.Response;
import jk.fox.config.transport.netty.NettyConfiguration;
import jk.fox.exchange.ExchangeChannel;
import jk.fox.exchange.netty.NettyChannel;
import jk.fox.transport.ClientTransporter;
import jk.fox.transport.netty.codec.HeartBeatDecode;
import jk.fox.transport.netty.codec.HeartBeatEncode;
import jk.fox.transport.netty.codec.RpcRequestEncode;
import jk.fox.transport.netty.codec.RpcResponseDecode;
import org.apache.log4j.Logger;

import java.net.SocketAddress;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;

/**
 * @author cuichao
 * @Description: NettyClientTransporter
 * @Date: create in 2020/3/28 10:59
 */
public class NettyClientTransporter implements ClientTransporter<ExchangeChannel<ChannelFuture>> {
    private static Logger logger = Logger.getLogger(NettyClientTransporter.class);

    private Bootstrap bootstrap;
    private MessageDispatcher dispatcher;


    private NettyConfiguration config;
    private boolean enableClientHeartBeat;
    private boolean keepAlive;
    private boolean tcp_noDelay;
    private int allIdleTimeSeconds;
    private int readerIdleTimeSeconds;
    private int writerIdleTimeSeconds;


    public NettyClientTransporter(NettyConfiguration config, MessageDispatcher dispatcher){
        this.config =  config;
        this.dispatcher = dispatcher;
        initWithConfig(this.config);
        getExecutors();
    }

    private void initWithConfig(NettyConfiguration config){
        this.enableClientHeartBeat = config.isEnableServerHeartBeat();
        this.keepAlive = config.isKeepAlive();
        this.tcp_noDelay = config.isDisableNagle();
        this.readerIdleTimeSeconds = config.getReaderIdleTimeSeconds();
        this.writerIdleTimeSeconds = config.getWriterIdleTimeSeconds();
        this.allIdleTimeSeconds = config.getAllIdleTimeSeconds();
    }

    /**
     * 得到客户端发送事件组
     * @return
     */
    public void getExecutors(){
        EventLoopGroup executors = new NioEventLoopGroup();
        bootstrap = new Bootstrap();
        bootstrap
                .group(executors)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY,tcp_noDelay)
                .option(ChannelOption.SO_KEEPALIVE, keepAlive)
                //.handler(new LoggingHandler(LogLevel.DEBUG))
                .attr(AttributeKey.valueOf(RpcDict.PING_COUNTER),0)
                .handler(channelInitializer());
    }

    /**
     * 创建channel-pipeline执行链
     * @return
     */
    public ChannelInitializer channelInitializer(){
        RpcClientChannelHandler handler = new RpcClientChannelHandler();
        return new ChannelInitializer() {
            @Override
            protected void initChannel(Channel ch) throws Exception {
                ch.pipeline().addLast(new HeartBeatEncode());
                ch.pipeline().addLast(new HeartBeatDecode());
                //消息编码器
                ch.pipeline().addLast(new RpcResponseDecode());
                //消息解码器
                ch.pipeline().addLast(new RpcRequestEncode());
                if(enableClientHeartBeat){
                    //心跳事件触发发器
                    ch.pipeline().addLast(new IdleStateHandler(readerIdleTimeSeconds,writerIdleTimeSeconds,allIdleTimeSeconds));
                }
                //心跳发送器
                ch.pipeline().addLast(new PingEmitter(dispatcher));
                //业务单元执行器
                ch.pipeline().addLast(handler);
            }
        };
    }

    @Override
    public Response send(ExchangeChannel<ChannelFuture> future, Request request,long timeout) throws Exception {
        Channel channel = future.channel().channel();
        //创建阻塞线程列队
        SynchronousQueue<Response> sync  = new SynchronousQueue<>();
        selector.put(request.getRequestId(),sync);
        //发送数据
        channel.writeAndFlush(request).addListener(f -> {
            if(f.isSuccess()){
                logger.info("request is send success. the requestId is "+request.getRequestId());
            }else {
                logger.error("request is send failure. the requestId is "+request.getRequestId(),f.cause());
            }

        });
        //阻塞 此处可以用sync.offer 设置超时时间 后期优化(已优化 配合timeout异常)
        try{
            Response response = sync.poll(timeout ,TimeUnit.MILLISECONDS);
            if(response == null){
                throw new RpcRequestTimeoutException(String.format("the requestId is %s of response is timeout",request.getRequestId()));
            }
            return response;
        }finally {
            //清除列队
            selector.remove(request.getRequestId());
        }
    }

    @Override
    public ExchangeChannel<ChannelFuture> doConnection(SocketAddress address) throws Exception {
        ChannelFuture sync = bootstrap.connect(address).sync();
        NettyChannel channel = new NettyChannel(sync);
        return channel;
    }



}
